package grabbitmq

import (
	"errors"
	"fmt"
	amqp "github.com/rabbitmq/amqp091-go"
	"go.uber.org/zap"
	"sync"
	"time"
	"wzgames/pkg/config"
	"wzgames/pkg/logger"
	"wzgames/pkg/utils/stringutil"
)

var (
	errMqNotFound = errors.New("RabbitMQ Not Found")

	rbMgr *rabbits
	once  = sync.Once{}

	StateClosed    = uint8(0)
	StateOpened    = uint8(1)
	StateReopening = uint8(2)
)

type Connection struct {
	*amqp.Connection
}

type rabbits struct {
	mqs map[string]*MQ

	mu sync.RWMutex
}

type MqConf struct {
	URIs []string
	Cfg  amqp.Config
}

type MQ struct {
	name   string
	conn   *Connection
	config *MqConf
	idx    int

	mu sync.RWMutex

	producers map[string]*Producer
	consumers map[string]*Consumer

	// connect error
	closeC chan *amqp.Error
	// closed by user
	stopC chan struct{}

	state uint8
}

func NewMQ(name string, confData config.ConfData) (*MQ, error) {
	if rbMgr == nil {
		once.Do(func() {
			rbMgr = &rabbits{
				mqs: make(map[string]*MQ),
			}
		})
	}
	return rbMgr.createDS(name, confData)
}

func (dm *rabbits) createDS(name string, confData config.ConfData) (*MQ, error) {
	dm.mu.Lock()
	defer dm.mu.Unlock()
	if confData == nil {
		return nil, fmt.Errorf("rabbitmq source=%s create Error on nil configor", name)
	}
	c := dm.parseConfig(confData)
	if c == nil {
		return nil, fmt.Errorf("rabbitmq source=%s create Error on parse config data", name)
	}
	if db, ok := dm.mqs[name]; ok {
		return db.UpdatePool(name, c)
	}
	return dm.create(name, c)
}

func (dm *rabbits) create(name string, conf *MqConf) (*MQ, error) {
	dm.mqs[name] = &MQ{
		name:      name,
		idx:       0,
		producers: make(map[string]*Producer),
		consumers: make(map[string]*Consumer),
		config:    conf,
		state:     StateClosed,
	}
	return dm.mqs[name], nil
}

func (dm *rabbits) parseConfig(confData config.ConfData) *MqConf {
	var uris []string
	confData.ForEachKey(func(key string, val interface{}) {
		uris = append(uris, stringutil.ToString(val))
	})

	return &MqConf{
		URIs: uris,
		Cfg:  amqp.Config{},
	}
}

func (dm *rabbits) Get(name string) *MQ {
	dm.mu.RLock()
	defer dm.mu.RUnlock()
	if p, ok := dm.mqs[name]; ok {
		return p
	}

	panic(errMqNotFound)
}

func (dm *rabbits) Destroy(name string) {
	dm.mu.Lock()
	defer dm.mu.Unlock()
	if v, ok := dm.mqs[name]; ok {
		err := v.Destroy()
		if err != nil {
			logger.Error("[RabbitMQ] rabbitmq destroy fail", zap.String("RabbitMQ", v.Name()))
		}
		delete(dm.mqs, name)
	}
}

func (dm *rabbits) DestroyAll() {
	dm.mu.RLock()
	defer dm.mu.RUnlock()
	for _, v := range dm.mqs {
		err := v.Destroy()
		if err != nil {
			logger.Error("[RabbitMQ] rabbitmq destroy fail", zap.String("RabbitMQ", v.Name()))
		}
	}
}

// MQ Methods
func (m *MQ) Name() string {
	return m.name
}

func (m *MQ) Open() (mq *MQ, err error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.state == StateOpened {
		return m, errors.New("MQ: Had been opened")
	}
	uri := m.config.URIs[m.idx]
	m.idx++
	if m.idx >= len(m.config.URIs) {
		m.idx = 0
	}

	conn, err := amqp.DialConfig(uri, m.config.Cfg)
	if err != nil {
		return nil, fmt.Errorf("MQ: Dial %s err: %v", uri, err)
	}

	m.conn = &Connection{
		Connection: conn,
	}

	m.state = StateOpened
	m.stopC = make(chan struct{})
	m.closeC = make(chan *amqp.Error, 1)
	m.conn.NotifyClose(m.closeC)

	go m.keepalive()

	return m, nil
}

func (m *MQ) keepalive() {
	select {
	case <-m.stopC:
		m.mu.Lock()
		m.conn.Close()
		m.state = StateClosed
		m.mu.Unlock()
	case err := <-m.closeC:
		if err == nil {
			logger.Error("[RabbitMQ] Disconnected with MQ")
		} else {
			logger.Error("[RabbitMQ] Disconnected with MQ", zap.Error(err))
		}

		// reconnect
		m.mu.Lock()
		m.state = StateReopening
		m.mu.Unlock()

		maxRetry := 99999999
		for i := 0; i < maxRetry; i++ {
			time.Sleep(time.Second)
			if _, e := m.Open(); e != nil {
				//glog.Error("Connection recover failed", log.Int("Tries", i+1), log.Err(e))
				continue
			}
			return
		}
		logger.Error("[RabbitMQ] Try to reconnect to MQ failed over maxRetry and exit.", zap.Int("MaxTries", maxRetry))
	}
}

func (m *MQ) UpdatePool(name string, conf *MqConf) (*MQ, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	m.config = conf

	return m, nil
}

func (m *MQ) ParseConfig(confData config.ConfData) *MqConf {
	return rbMgr.parseConfig(confData)
}

func (m *MQ) State() uint8 {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.state
}

func (m *MQ) Producer(name string) (*Producer, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.state != StateOpened {
		return nil, fmt.Errorf("MQ: Not initialized, now state is %d", m.State)
	}
	p, ok := m.producers[name]
	if !ok {
		p = newProducer(name, m)
		m.producers[name] = p
	}

	return p, nil
}

func (m *MQ) CloseProducer(name string) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.state != StateOpened {
		return fmt.Errorf("MQ: Not initialized, now state is %b", m.State())
	}
	if p, ok := m.producers[name]; ok {
		p.Close()
		return nil
	}
	return errors.New("MQ: producer not exist")
}

func (m *MQ) Consumer(name string) (*Consumer, error) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if m.state != StateOpened {
		return nil, fmt.Errorf("MQ: Not initialized, now state is %d", m.State)
	}
	consumer, ok := m.consumers[name]
	if !ok {
		consumer = newConsumer(name, m)
		m.consumers[name] = consumer
	}

	return consumer, nil
}

func (m *MQ) CloseConsumer(name string) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.state != StateOpened {
		return fmt.Errorf("MQ: Not initialized, now state is %b", m.State())
	}
	if consumer, ok := m.consumers[name]; ok {
		consumer.Close()
		return nil
	}

	return errors.New("MQ: producer not exist")
}

func (m *MQ) channel() (*amqp.Channel, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()
	return m.conn.Channel()
}

func (m *MQ) Close() {
	// close producers
	for _, p := range m.producers {
		p.Close()
	}
	m.producers = make(map[string]*Producer)

	// close consumers
	for _, c := range m.consumers {
		c.Close()
	}
	m.consumers = make(map[string]*Consumer)

	// close mq connection
	select {
	case <-m.stopC:
		// had been closed
	default:
		close(m.stopC)
	}

	logger.Info("[RabbitMQ] MQ Closed", zap.String("MQ", m.Name()))

	// wait done
	for m.State() != StateClosed {
		time.Sleep(time.Second)
	}
}

func (m *MQ) Destroy() error {
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.state == StateClosed {
		return nil
	}
	m.Close()

	return nil
}
